package me.zhengjie.stock.pipeline;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import me.zhengjie.stock.domain.TblKline;
import me.zhengjie.stock.domain.TblStockInfo;
import me.zhengjie.stock.domain.TblStockPools;
import me.zhengjie.stock.domain.TblTradingDateRecord;
import me.zhengjie.stock.service.TblKlineService;
import me.zhengjie.stock.service.TblStockInfoService;
import me.zhengjie.stock.service.TblStockPoolsService;
import me.zhengjie.stock.service.TblTradingDateRecordService;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import us.codecraft.webmagic.Page;
import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Spider;
import us.codecraft.webmagic.Task;
import us.codecraft.webmagic.pipeline.Pipeline;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 个股实时行情
 *
 * @author Aaron
 * @date 2020/12/20
 */
@Slf4j
@Component
@AllArgsConstructor
public class StockRealTimeSpider {

    private final TblStockInfoService stockInfoService;
    private final TblKlineService klineService;
    private final TblStockPoolsService stockPoolsService;
    private final TblTradingDateRecordService tblTradingDateRecordService;

    /**
     * 监测今日交易状态 并 更新交易日记录
     */
    public void runTradingDateKlineSpider(String tradingDate, String startCode, String endCode) {
        log.info("------------ start runTradingDateKlineSpider ... --------------");
        try {
            LambdaQueryWrapper<TblStockInfo> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(TblStockInfo::getOpenStatus, 1);
            queryWrapper.ge(StringUtils.isNotEmpty(startCode), TblStockInfo::getCode, startCode);
            queryWrapper.le(StringUtils.isNotEmpty(endCode), TblStockInfo::getCode, endCode);
            queryWrapper.orderByAsc(TblStockInfo::getCode);
            List<TblStockInfo> stockInfoList = stockInfoService.list(queryWrapper);
            AtomicInteger processed = new AtomicInteger();
            stockInfoList.stream().forEach(stockInfo -> {
                processed.addAndGet(1);
                String url = getUrl(tradingDate, stockInfo.getCode(), stockInfo.getMarket());
                Spider.create(new StockRealTimePageProcessor()).addUrl(url).addPipeline(new UpdateTradingDatePipeline()).run();
            });
            log.info("----------- runKlineSpider finished! ---------------");
            log.info("----------- runKlineSpider finished! processed:{} ---------------", processed);
        } catch (Exception e) {
            log.error("runTradingDateKlineSpider error:{}", e);
        }
    }

    public void runTradingDateKlineSpiderForLongLet(String tradingDate) {
        log.info("------------ start runKlineSpider ... --------------");
        try {
            LambdaQueryWrapper<TblStockPools> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.orderByAsc(TblStockPools::getCode);
            List<TblStockPools> stockInfoList = stockPoolsService.list(queryWrapper);
            stockInfoList.stream().forEach(stockInfo -> {
                String url = getUrl(tradingDate, stockInfo.getCode(), getMarket(stockInfo.getCode()));
                Spider.create(new StockRealTimePageProcessor()).addUrl(url).addPipeline(new UpdateTradingDatePipeline()).run();
            });
        } catch (Exception e) {
            log.error("runTradingDateKlineSpiderForLongLet error:{}", e);
        }
    }

    /**
     * 拼接url
     *
     * @param code
     * @param market
     * @return
     */
    private String getUrl(String tradingDate, String code, Integer market) {
        String secid = market + "." + code;
        String fields = "f43,f44,f45,f46,f47,f48,f50,f51,f52,f57,f58,f60,f86,f127,f128,f168,f169,f170,f292";
        String url = "http://push2.eastmoney.com/api/qt/stock/get" +
                "?ut=fa5fd1943c7b386f172d6893dbfba10b" +
                "&invt=2&fltt=2" +
                "&fields=" + fields +
                "&secid=" + secid +
                "&cb=jQuery112403418241714388266_" + tradingDate +
                "&_=" + tradingDate;
        return url;
    }

    /**
     * 解析数据
     * f43: 26.9,	实时价格
     * f44: 27.23,	最高
     * f45: 26.72,	最底
     * f46: 27.09,	今开
     * f47: 1672064,		成交量
     * f48: 4509427200,	成交额
     * f50: 0.81,	量比
     * f51: 32.77,	涨停
     * f52: 21.85,	跌停
     * f57: "300059",
     * f58: "东方财富",
     * f60: 27.31,		昨收
     * f86: 1608276843,	时间戳（秒）
     * f127: "电子信息",
     * f128: "上海板块",
     * f292: 5,			股票状态，2 交易中，5 已收盘
     * f168: "换手率",
     * f169: "涨跌额",
     * f170	 "涨跌幅"
     */
    private class StockRealTimePageProcessor extends MyPageProcessor {
        @Override
        public void process(Page page) {
            String pageString = page.getRawText();
            int startIndex = pageString.indexOf("(");
            int endIndex = pageString.lastIndexOf(")");
            String jsonString = pageString.substring(startIndex + 1, endIndex);
            JSONObject pageJson = JSONObject.parseObject(jsonString);
            JSONObject dataJson = pageJson.getJSONObject("data");
            page.putField("data", dataJson);
        }
    }


    private class UpdateTradingDatePipeline implements Pipeline {
        @Override
        public void process(ResultItems resultItems, Task task) {
            JSONObject dataJson = resultItems.get("data");
            TblKline tblKlineNew = toObject(dataJson);
//            log.info("StockRealTimeSpider{}", JSON.toJSONString(dataJson));
            LambdaQueryWrapper<TblStockInfo> updateWrapper = new LambdaQueryWrapper<>();
            updateWrapper.eq(TblStockInfo::getCode, tblKlineNew.getCode());
            updateWrapper.eq(TblStockInfo::getName, tblKlineNew.getName());
            TblStockInfo stockInfoDb = new TblStockInfo();
            stockInfoDb.setCity(dataJson.getString("f128"));
            stockInfoDb.setIndustry(dataJson.getString("f127"));
            stockInfoDb.setGmtModified(new Date());
            stockInfoService.update(stockInfoDb, updateWrapper);
            //判断交易数据是否存在
            LambdaQueryWrapper<TblKline> queryWrapper = new LambdaQueryWrapper<>();
            queryWrapper.eq(TblKline::getCode, tblKlineNew.getCode());
            queryWrapper.eq(TblKline::getName, tblKlineNew.getName());
            queryWrapper.eq(TblKline::getTradingDate, tblKlineNew.getTradingDate());
            TblKline tblKline = klineService.getOne(queryWrapper);
            if (tblKline != null) {
                klineService.update(tblKlineNew, queryWrapper);
            } else {
                klineService.save(tblKlineNew);
            }
        }
    }

    private TblKline toObject(JSONObject dataJson) {
        TblKline kline = new TblKline();
        kline.setCode(dataJson.getString("f57"));
        kline.setName(dataJson.getString("f58"));
        Long time = dataJson.getLong("f86");
        Calendar calendar = new Calendar.Builder().setInstant(time * 1000).build();
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        String tradingDate = sdf.format(calendar.getTime());
        kline.setTradingDate(tradingDate);
        kline.setOpeningPrice(dataJson.getBigDecimal("f46"));
        //实时价格
        kline.setClosingPrice(dataJson.getBigDecimal("f43"));
        kline.setPeakPrice(dataJson.getBigDecimal("f44"));
        kline.setBottomPrice(dataJson.getBigDecimal("f45"));
        kline.setTradingVolume(dataJson.getInteger("f47"));
        kline.setTradingAmount(dataJson.getBigDecimal("f48"));
        //振幅
//        kline.setAmplitudeRate(dataJson.getBigDecimal("??"));
        //换手率
        kline.setTurnoverRate(dataJson.getBigDecimal("f168"));
        //涨跌额
        kline.setChangeAmount(dataJson.getBigDecimal("f169"));
        //涨跌幅
        kline.setChangeRate(dataJson.getBigDecimal("f170"));
        kline.setGmtModified(new Date());
        kline.setGmtCreate(new Date());
        //"f128":"广东板块","f127":"银行"
        return kline;
    }


    /**
     * 监测今日交易状态 并 更新交易日记录
     */
    public void checkAndInitTradingDate() {
        String code = "000001";
        String tradingDate = DateUtil.formatDate(new Date());
        String url = getUrl(tradingDate, code, getMarket(code));
        Spider.create(new StockRealTimePageProcessor()).addUrl(url).addPipeline(new InitTradingDatePipeline()).run();
    }

    /**
     * 初始化交易日期记录
     */
    private class InitTradingDatePipeline implements Pipeline {
        @Override
        public void process(ResultItems resultItems, Task task) {
            JSONObject dataJson = resultItems.get("data");
            Boolean isOpen = "2".equals(dataJson.getString("f292"));
            if (isOpen) {
                Long time = dataJson.getLong("f86");
                Calendar calendar = new Calendar.Builder().setInstant(time * 1000).build();
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                String tradingDate = sdf.format(calendar.getTime());
                String todayDate = DateUtil.formatDate(new Date());
                if (todayDate.equalsIgnoreCase(tradingDate)) {
                    TblTradingDateRecord record = new TblTradingDateRecord();
                    record.setTradingDate(todayDate);
                    record.setGmtModified(new Date());
                    record.setGmtCreate(new Date());
                    // 检测到今日开市状态
                    LambdaQueryWrapper<TblTradingDateRecord> queryWrapper = new QueryWrapper<TblTradingDateRecord>().lambda();
                    queryWrapper.eq(TblTradingDateRecord::getTradingDate, todayDate);
                    TblTradingDateRecord tradingDateRecord = tblTradingDateRecordService.getOne(queryWrapper);
                    if (tradingDateRecord == null) {
                        tblTradingDateRecordService.save(record);
                        //保存数据
                        log.info("UpdateTradingDatePipeline success! Today is Trading date!");
                    } else {
                        log.info("UpdateTradingDatePipeline success! Today is Trading date! TradingDateData is exit!!");
                    }
                }
                log.info("StockRealTimeSpider{}", JSON.toJSONString(dataJson));
            } else {
                log.info("UpdateTradingDatePipeline success! Today is not Trading date!");
            }
        }
    }

    private Integer getMarket(String code) {
        //市场，0 深证，1 上证
        return code.startsWith("6") ? 1 : 0;
    }
}



